package com.zhanghe.study;

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.kstream.KTable;

import java.util.Properties;
import java.util.concurrent.CountDownLatch;

public class KTableTest {
    public static void main(String[] args) {
        Properties prop = KStreamTest.init();

        StreamsBuilder builder = new StreamsBuilder();
        // 构造kstream日志流（数据来源来自  kafka的主题stream-input）
        KTable<String,String> kTable = builder.table("stream-input");
        // 打印到控制台

        Topology topology = builder.build();
        System.out.println(topology.describe());
        final KafkaStreams kafkaStreams = new KafkaStreams(topology,prop);
        final CountDownLatch latch = new CountDownLatch(1);
        Runtime.getRuntime().addShutdownHook(new Thread("hook"){
            @Override
            public void run() {
                System.out.println("线程停止");
                kafkaStreams.close();
                latch.countDown();
            }
        });
        try {
            kafkaStreams.start();
            latch.await();
        } catch (InterruptedException e){
            System.exit(1);
        }
        System.out.println("项目终止");
        System.exit(0);
    }
}
